package com.xj.driver.mqtt.subscriber.monitor;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicLong;

/**
 * 消息积压监控器
 */
@Slf4j
@Component
@Data
public class MessageBacklogMonitor {

    private final AtomicLong totalReceived = new AtomicLong(0);
    private final AtomicLong totalProcessed = new AtomicLong(0);
    private final AtomicLong totalDiscarded = new AtomicLong(0);
    private final AtomicLong queueSize = new AtomicLong(0);

    // 积压阈值
    private static final long BACKLOG_THRESHOLD = 5000;
    private static final long CRITICAL_BACKLOG_THRESHOLD = 8000;

    /**
     * 记录消息接收
     */
    public void recordMessageReceived() {
        totalReceived.incrementAndGet();
        queueSize.incrementAndGet();
        checkBacklogWarning();
    }

    /**
     * 记录消息处理完成
     */
    public void recordMessageProcessed() {
        totalProcessed.incrementAndGet();
        queueSize.decrementAndGet();
    }

    /**
     * 记录消息丢弃
     */
    public void recordMessageDiscarded() {
        totalDiscarded.incrementAndGet();
        queueSize.decrementAndGet();
    }

    /**
     * 获取当前积压数量
     */
    public long getCurrentBacklog() {
        return queueSize.get();
    }

    /**
     * 检查积压警告
     */
    private void checkBacklogWarning() {
        long backlog = getCurrentBacklog();
        if (backlog >= CRITICAL_BACKLOG_THRESHOLD) {
            log.error("🚨 消息积压严重! 当前积压数量: {}", backlog);
        } else if (backlog >= BACKLOG_THRESHOLD) {
            log.warn("⚠️ 消息积压警告! 当前积压数量: {}", backlog);
        }
    }

    /**
     * 获取监控统计
     */
    public BacklogStats getStats() {
        return new BacklogStats(
                totalReceived.get(),
                totalProcessed.get(),
                totalDiscarded.get(),
                queueSize.get(),
                System.currentTimeMillis()
        );
    }

    @Data
    public static class BacklogStats {
        private final long totalReceived;
        private final long totalProcessed;
        private final long totalDiscarded;
        private final long currentBacklog;
        private final long timestamp;

        public double getProcessRate() {
            return totalReceived > 0 ? (double) totalProcessed / totalReceived * 100 : 0;
        }

        public double getDiscardRate() {
            return totalReceived > 0 ? (double) totalDiscarded / totalReceived * 100 : 0;
        }
    }
}